package com.heima.schedule.service.impl;

import com.alibaba.fastjson.JSON;
import com.heima.common.cache.CacheService;
import com.heima.model.schedule.dto.Task;
import com.heima.model.schedule.pojos.Taskinfo;
import com.heima.model.schedule.pojos.TaskinfoLogs;
import com.heima.schedule.service.TaskService;
import com.heima.schedule.service.TaskinfoLogsService;
import com.heima.schedule.service.TaskinfoService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.util.Date;
import java.util.List;
import java.util.Set;

@Slf4j
@Service
@Transactional
public class TaskServiceImpl implements TaskService {
    @Autowired
    private CacheService cacheService;

    @Autowired
    private TaskinfoService taskinfoService;

    @Autowired
    private TaskinfoLogsService taskinfoLogsService;

    //添加任务
    @Override
    @Transactional(rollbackFor = RuntimeException.class)
    public Long addTask(Task task) {
        //0.入参判空
        if (task == null) {
            log.warn("入参为空");
            return null;
        }

        //1.插入taskInfo表
        Taskinfo taskinfo = new Taskinfo();
        BeanUtils.copyProperties(task, taskinfo);
        taskinfo.setExecuteTime(new Date(task.getExecuteTime()));
        boolean taskinfoResponse = taskinfoService.save(taskinfo);
        if (!taskinfoResponse) {
            log.warn("taskInfo插入失败");
            return null;
        }

        //2.插入taskInfoLogs表
        TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
        BeanUtils.copyProperties(task, taskinfoLogs);
        taskinfoLogs.setExecuteTime(new Date(task.getExecuteTime()));
        taskinfoLogs.setStatus(0);//初始化
        boolean taskinfoLogsResponse = taskinfoLogsService.save(taskinfoLogs);
        if (!taskinfoLogsResponse) {
            log.warn("taskInfoLogs日志插入失败");
            throw new RuntimeException("taskInfoLogs日志插入失败");
        }

        long executeTime = task.getExecuteTime();//任务时间
        long currentTime = System.currentTimeMillis();//当前时间
        long futureTime = currentTime + (5 * 60 * 1000);//未来五分钟
        //3.判断任务执行时间 是否小于 当前时间  是:执行任务
        if (executeTime < currentTime) {
            //执行任务,进入任务队列
            cacheService.lLeftPush("TOPIC", JSON.toJSONString(task));
        }

        //4.判断任务执行时间 是否小于 当前时间+5min  任务执行时间(还有五分钟任务开始)
        if (executeTime < futureTime) {
            //执行任务,进入未来队列  队列,任务,分值
            cacheService.zAdd("FUTURE", JSON.toJSONString(task), task.getExecuteTime());
        }

        //5.返回任务Id
        return task.getTaskId();
    }

    //取消任务
    @Override
    @Transactional(rollbackFor = RuntimeException.class)
    public Boolean cancelTask(Long taskId) {
        //0.入参判空
        if (taskId == null) {
            log.warn("入参为空");
            return false;
        }

        //1.删除taskInfo表
        boolean taskinfoResponse = taskinfoService.removeById(taskId);
        if (!taskinfoResponse) {
            log.warn("任务删除失败");
            return false;
        }

        //2.修改taskInfoLogs表
        TaskinfoLogs taskinfoLogs = taskinfoLogsService.getById(taskId);
        taskinfoLogs.setStatus(2);//修改任务状态为取消
        boolean taskinfoLogsResponse = taskinfoLogsService.updateById(taskinfoLogs);
        if (!taskinfoLogsResponse) {
            log.warn("任务日志状态修改失败");
            throw new RuntimeException("任务日志状态修改失败");
        }

        //3.组装task
        Task task = new Task();
        BeanUtils.copyProperties(taskinfoLogs, task);
        task.setExecuteTime(taskinfoLogs.getExecuteTime().getTime());//修改时间格式
        String taskString = JSON.toJSONString(task);

        //4.删除redis中list 和 zset 结构数据
        Long removeResponse = cacheService.lRemove("TOPIC", 0, taskString);

        Long zRemoveResponse = cacheService.zRemove("FUTURE", taskString);

        return true;
    }

    //消费任务
    @Override
    @Transactional(rollbackFor = RuntimeException.class)
    public Task pollTask() {

        //1.从TOPIC队列中pop出task
        String pop = cacheService.lRightPop("TOPIC");
        if (StringUtils.isEmpty(pop)) {
            log.warn("没有可执行任务");
            return null;
        }
        Task task = JSON.parseObject(pop, Task.class);//json转换
        if (task == null) {
            log.warn("没有可执行任务");
            return null;
        }

        //2.删除taskinfo表中任务
        boolean removeResponse = taskinfoService.removeById(task.getTaskId());
        if (!removeResponse) {
            log.warn("任务删除失败");
            return null;
        }

        //3.修改taskinfoLogs表中状态
        TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
        BeanUtils.copyProperties(task, taskinfoLogs);
        taskinfoLogs.setExecuteTime(new Date(task.getExecuteTime()));
        taskinfoLogs.setStatus(1);//executed
        boolean updateResponse = taskinfoLogsService.updateById(taskinfoLogs);
        if (!updateResponse) {
            log.warn("任务日志修改失败");
            throw new RuntimeException("任务日志修改失败");
        }
        return task;
    }

    /**
     * 定时刷新任务
     */
    @Scheduled(cron = "0 */1 * * * ?")//每分钟执行一次
    public void refresh() {
        System.out.println("执行定时刷新任务");

        //1.从FUTURE队列中取到任务
        // 获取该组key下当前需要消费的任务数据
        // 获取0~当前时间的所有任务（代表小于当前时间的任务）
        // 任务a，执行时间100000,(0 ~ 100020)   转移到执行队列
        // 任务b，执行时间100300,(0 ~ 100020)   不在0-当前时间,还没到任务执行时间,不用转移
        Set<String> tasks = cacheService.zRangeByScore("FUTURE", 0, System.currentTimeMillis());
        if (!tasks.isEmpty()) {
//            log.warn("没有要添加到执行队列的任务");
//            return;
            //2.将这些任务添加到消费队列中  参数1:来源的队列  参数2:目标队列  参数3:转移的数据
            cacheService.refreshWithPipeline("FUTURE", "TOPIC", tasks);//将这些任务数据添加到消费者队列中
            System.out.println("成功的将FUTURE下的当前需要执行的任务数据刷新到TOPIC下");
        }

    }

    //初始化
    @Scheduled(fixedDelay = 5 * 60 * 1000)//5分钟刷新一次
    public void init() {
        //1.清空redis缓存
        cacheService.delete("TOPIC");//清空执行队列
        cacheService.delete("FUTURE");//清空未来队列

        //2.从数据库中查询Taskinfo表拿出任务
        List<Taskinfo> list = taskinfoService.list();
        if (CollectionUtils.isEmpty(list)) {
            log.warn("没有任务");
            return;
        }

        //3.保存到redis中
        for (Taskinfo taskinfo : list) {
            if (taskinfo == null) {
                continue;
            }
            long executeTime = taskinfo.getExecuteTime().getTime();
            long currentTime = System.currentTimeMillis();
            long futureTime = currentTime + (5 * 60 * 1000);

            //4.组装json类型的task
            Task task = new Task();
            BeanUtils.copyProperties(taskinfo, task);
            task.setExecuteTime(executeTime);
            String taskString = JSON.toJSONString(task);

            //5.判断任务执行时间 是否小于 当前时间  是:执行任务
            if (executeTime < currentTime) {
                //执行任务,进入任务队列
                cacheService.lLeftPush("TOPIC", taskString);
            }
            //6.判断任务执行时间 是否小于 当前时间+5min  任务执行时间(还有五分钟任务开始)
            if (executeTime < futureTime) {
                //执行任务,进入未来队列  队列,任务,分值
                cacheService.zAdd("FUTURE", taskString, task.getExecuteTime());
            }
        }
    }
}
